package org.qnit.dw

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.qnit.util.SparkUtil

object DwsApp {

  def main(args: Array[String]): Unit = {
    val env = args(0)
    println(s"env=$env")
    if (!SparkUtil.verifyEnv(env))
      System.exit(1)
    val spark: SparkSession = SparkUtil.initSparkSession(env, "DwsApp")

    println("---------- dws.dws_user_cnt ----------")
    spark.sql(
      s"""
         |select province,
         |       if(city is null, 'TOTAL', city) city,
         |       if(county is null, 'TOTAL', county) county,
         |       count(*) cnt
         |from ods.ods_user_info
         |where province > ''
         |group by province, city, county
         |grouping sets (
         |    (province),
         |    (province, city),
         |    (province, city, county)
         |)
         |order by 1, 2, 3
         |""".stripMargin).repartition(1)
      .write.mode(SaveMode.Overwrite)
      .format("parquet")
      .saveAsTable(s"dws.dws_user_cnt")

    println("---------- dws.dws_expert_logbook_cnt ----------")
    spark.sql(
      s"""
         |select province_name,
         |       if(city_name is null, 'TOTAL', city_name) city_name,
         |       if(county_name is null, 'TOTAL', county_name) county_name,
         |       count(*) cnt,
         |       count(distinct user_id) uv
         |from dwd.dwd_expert_logbook_info
         |group by province_name, city_name, county_name
         |grouping sets (
         |    (province_name),
         |    (province_name, city_name),
         |    (province_name, city_name, county_name)
         |)
         |order by 1, 2, 3
         |""".stripMargin).repartition(1)
      .write.mode(SaveMode.Overwrite)
      .format("parquet")
      .saveAsTable(s"dws.dws_expert_logbook_cnt")

    println("---------- dws.dws_expert_logbook_yymm_cnt ----------")
    spark.sql(
      s"""
         |select year_month, province_name,
         |       if(city_name is null, 'TOTAL', city_name) city_name,
         |       if(county_name is null, 'TOTAL', county_name) county_name,
         |       count(*) cnt,
         |       count(distinct user_id) uv
         |from (
         |    select v0.*, substring(gmt_create, 1, 7) year_month
         |    from dwd.dwd_expert_logbook_info v0
         |) v1
         |group by year_month, province_name, city_name, county_name
         |grouping sets (
         |    (year_month, province_name),
         |    (year_month, province_name, city_name),
         |    (year_month, province_name, city_name, county_name)
         |)
         |order by 1, 2, 3, 4
         |""".stripMargin).repartition(1)
      .write.mode(SaveMode.Overwrite)
      .format("parquet")
      .saveAsTable(s"dws.dws_expert_logbook_yymm_cnt")

    println("---------- dws.dws_label_cnt ----------")
    spark.sql(
      s"""
         |select row_number() over(order by uv desc) rank, label, uv
         |from (
         |    select label, count(*) uv
         |    from (
         |        select user_id, explode(split(top_term, ',')) label
         |        from dwb.dwb_user_label
         |    ) v1
         |    group by 1
         |) v2
         |""".stripMargin).repartition(1)
      .write.mode(SaveMode.Overwrite)
      .format("parquet")
      .saveAsTable(s"dws.dws_label_cnt")

    println("---------- dws.dws_area_label ----------")
    spark.sql(
      s"""
         |select province_name, 'TOTAL' city_name, 'TOTAL' county_name, c as label,
         |       row_number() over(partition by province_name order by seq desc) score
         |from dwb.dwb_province_label
         |lateral view posexplode(split(top_term, ',')) v1 as seq, c
         |union all
         |select province_name, city_name, 'TOTAL' county_name, c as label,
         |       row_number() over(partition by province_name, city_name order by seq desc) score
         |from dwb.dwb_city_label
         |lateral view posexplode(split(top_term, ',')) v1 as seq, c
         |union all
         |select province_name, city_name, county_name, c as label,
         |       row_number() over(partition by province_name, city_name, county_name order by seq desc) score
         |from dwb.dwb_county_label
         |lateral view posexplode(split(top_term, ',')) v1 as seq, c
         |""".stripMargin).repartition(1)
      .write.mode(SaveMode.Overwrite)
      .format("parquet")
      .saveAsTable(s"dws.dws_area_label")

    spark.stop()
  }

}
